package com.sg.java.apps;

import com.sg.java.util.SqlUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class RunData {

    private static final Logger log = LoggerFactory.getLogger(RunData.class);

    public static void main(String[] args) throws Exception {
        final File file = new File("C:\\Users\\HUAWEI\\Desktop\\volt");
        final File[] csvs = file.listFiles(pathname -> pathname.getName().endsWith(".csv"));
        if (csvs == null) {
            throw new RuntimeException("目录下无文件");
        }
        String schemaSql = "SELECT\n" +
                           "\tA.attname AS NAME,\n" +
                           "\tformat_type(A.atttypid, A.atttypmod) AS TYPE,\n" +
                           "\tA.attnotnull AS NOTNULL,\n" +
                           "\tcol_description(A.attrelid, A.attnum) AS COMMENT\n" +
                           "FROM\n" +
                           "\tpg_class AS C,\n" +
                           "\tpg_attribute AS A \n" +
                           "WHERE\n" +
                           "\tC.relname = '%s' \n" +
                           "\tAND A.attnum > 0\n" +
                           "\tAND A.attrelid = C.oid";

        Map<String, Long> infos = new ConcurrentHashMap<>(csvs.length);

        final Thread thread = new Thread(() -> {
            try {
                while (true) {
                    Thread.sleep(15000);
                    StringJoiner sj = new StringJoiner("\n");
                    infos.forEach((k, v) -> sj.add(k + ":" + v));
                    log.info("\n" + sj + "\n================================================================");
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        thread.setDaemon(true);
        thread.start();

        Arrays.stream(csvs).parallel().forEach(
                csv -> {
                    try {
                        int batchSize;
                        if (csv.length() <= 1024 * 1024 * 1024) {
                            batchSize = 5000;
                        } else if (csv.length() <= 1024L * 1024 * 1024 * 10) {
                            batchSize = 50000;
                        } else {
                            batchSize = 100000;
                        }
                        final Connection conn = DriverManager.getConnection("jdbc:postgresql://piscn.vicp.cc:55432/volt?", "volt", "volt2022");
                        final String tn = csv.getName().replaceAll(".csv", "");
                        String preSql = "insert into dwd_cms." + tn + " values ";
                        ResultSet rs = conn.prepareStatement(String.format(schemaSql, tn)).executeQuery();
                        Map<Integer, String> types = new HashMap<>();
                        int i = 0;
                        infos.put(tn, 0L);
                        while (rs.next()) {
                            final String type = rs.getString("type");
                            if (type.contains("text") || type.contains("varchar") || type.contains("char")) {
                                types.put(i++, "S");
                            } else if (type.contains("timestamp") || type.contains("time") || type.contains("date")) {
                                types.put(i++, "D");
                            } else {
                                types.put(i++, "N");
                            }
                        }
                        BufferedReader br = new BufferedReader(new FileReader(csv));
                        String line;
                        List<String> records = new ArrayList<>(batchSize);
                        while (true) {
                            line = br.readLine();
                            if (line == null) {
                                infos.put(tn, exe(types, records, preSql, conn) + infos.get(tn));
                                break;
                            } else {
                                if (line.startsWith("\"") && !line.endsWith("\"")) {
                                    line = line.replaceAll("\"", "");
                                    String nextLine;
                                    StringBuilder sb = new StringBuilder();
                                    while ((nextLine = br.readLine()) != null && !nextLine.endsWith("\"")) {
                                        sb.append(nextLine);
                                    }
                                    if (nextLine != null) {
                                        sb.append(nextLine.replaceAll("\"", ""));
                                    }
                                    line += sb.toString();
                                } else if (line.startsWith("\"") && line.endsWith("\"")) {
                                    line = line.replaceAll("\"", "");
                                }
                                records.add(line);
                            }
                            if (records.size() == batchSize) {
                                infos.put(tn, exe(types, records, preSql, conn) + infos.get(tn));
                            }
                        }
                        conn.close();
                        br.close();
                        System.out.println(tn + " over,count:" + infos.get(tn));
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
        );
    }

    public static long exe(Map<Integer, String> types, List<String> records, String preSql, Connection conn) {
        StringJoiner values = new StringJoiner(",");
        for (String r : records) {
            String[] split = r.split("!@@#", -1);
            StringJoiner value = new StringJoiner(",", "(", ")");
            for (int j = 0; j < split.length; j++) {
                final String v = split[j];
                String type = types.get(j);
                switch (type) {
                    case "N":
                        value.add(SqlUtils.toSqlValue(v, false));
                        break;
                    case "S":
                    case "D":
                        value.add(SqlUtils.toSqlValue(v, true));
                        break;
                }
            }
            values.add(value.toString());
        }
        records.clear();
        String finalSql = preSql + values;
        PreparedStatement ps;
        try {
            ps = conn.prepareStatement(finalSql);
            final int a = ps.executeUpdate();
            records.clear();
            return a;
        } catch (SQLException e) {
//            System.out.println(finalSql);
            throw new RuntimeException(e);
        }

    }

}
